package com.xzc.kafka.action.consumer._4Single;

public class Main {

    //offset 维护困难，每个worker还要维护offset，而且这个offset还要同步，线程会切换
    //如果worker挂了会丢失消息，这个没懂是为什么，worker挂了，那么他那一部分的offset就没更改
    //既然不更改的话，那么就不会提交，只有可能重复消费，不会丢消息啊。因为是at least once
    public static void main(String[] args) {
        String brokerList = "es1:19092,es2:19092,es3:19092";
        String topicName = "test-topic1";
        String groupID = "test-group";
        final ConsumerThreadHandler<byte[], byte[]> handler = new ConsumerThreadHandler<>(brokerList, groupID, topicName);
        final int cpuCount = Runtime.getRuntime().availableProcessors();

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                handler.consume(cpuCount);
            }
        };

        new Thread(runnable).start();

        try {
            Thread.sleep(20000L);
        } catch (InterruptedException e) {

        }
    }
}
